package com.huatai.bi.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huatai.bi.constant.KafkaConstant;
import com.huatai.bi.entity.MqMqttClientEntity;
import com.huatai.bi.service.MqMqttClientService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.PostConstruct;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @program: itc-airport-cloud
 * @description: mqtt回调
 * @author: songxiude
 * @create: 2024-07-15 15:20
 **/
@Slf4j
@Component
public class MqttClientCallback implements MqttCallback, MqttCallbackExtended {

    /**
     * 系统的mqtt客户端id
     */
    private String mqttClientId;

    @Autowired
    private MqttClientConnect mqttClientConnect;

    private static MqttClientCallback callback;

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;


    @Autowired
    private MqMqttClientService mqttClientService;

    @PostConstruct
    public void init() {
        callback = this;
    }

    public MqttClientCallback(String mqttClientId) {
        this.mqttClientId = mqttClientId;
    }

    public MqttClientCallback() {

    }

    /**
     * MQTT 断开连接会执行此方法
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.error("断开连接客户端id：{}",mqttClientId);
		log.error(throwable.getMessage(), throwable);
        //如果断开修改链接状态
        callback.mqttClientService.update(null,Wrappers.<MqMqttClientEntity>lambdaUpdate()
                .eq(MqMqttClientEntity::getClientId,mqttClientId).set(MqMqttClientEntity::getStatus,0));
    }

    /**
     * publish发布成功后会执行到这里
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        log.info("发布消息成功");
    }

    /**
     * subscribe订阅后得到的消息会执行到这里
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.info("收到来自 " + mqttClientId + "====" + topic + " 的消息：{}", new String(message.getPayload()));
        if (message.isRetained()) {
            log.info("收到保留消息：{}，消息质量：{}，消息主题：{}", message.getPayload(), message.getQos(), topic);
        }
        //ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
        String payload = message.toString();
        JSONObject jsonObject=new JSONObject();
        jsonObject.put("topic",topic);
        jsonObject.put("clientId",mqttClientId);
        try {
			jsonObject.put("data", JSONObject.parseObject(payload,Feature.OrderedField));
		}catch (Exception e){
			log.info("消息数据解析异常：{}",e.getMessage());
			jsonObject.put("data", payload);
			e.printStackTrace();
		}
        String jsonString = jsonObject.toJSONString();
        //推给kafka处理
        callback.kafkaTemplate.send(KafkaConstant.MQTT_DATA_ACCEPT,jsonString);
    }

    /**
     * 连接成功(首次连接、断开连接都会执行该方法)
     *
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
        MqttClientConnect mqttConnect = mqttClients.get(mqttClientId);

        if (ObjectUtils.isEmpty(mqttConnect)) {
            // 首次连接
            log.info("连接成功，clientId-->" + mqttClientId);
            return;
        }
        // 方法二：异常断开后重连，需要重新订阅主题(因为异常断开重连后不会订阅原来主题)
       //MqMqttClientService clientService = BeanUtil.getBean(MqMqttClientService.class);
        MqMqttClientEntity mqtt = callback.mqttClientService.getOne(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, mqttClientId));

        String topic = mqtt.getClientTopic();
        try {
            mqttConnect.sub(topic, 0);

			//订阅成功以后，修改客户端状态
			callback.mqttClientService.update(null,Wrappers.<MqMqttClientEntity>lambdaUpdate()
				.eq(MqMqttClientEntity::getClientId,mqttClientId).set(MqMqttClientEntity::getStatus,1));
			log.info("重新连接后，订阅成功，topic-->" + topic);
            //mqttConnect.sub(topic, 1);
        } catch (MqttException e) {
            e.printStackTrace();
            log.error("重新连接后订阅异常，topic-->" + topic);
        }
    }
}
